package org.terramagnet.mvcframework.asynchronous;

import org.terramagnet.mvcframework.persistent.HashFilePersistent;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * 异步单线程队列操作抽象类. 每个实例都会有个单独的线程。
 *
 * <p>
 * 系统正常停止时不会丢失数据。</p>
 *
 * @author lip
 */
public abstract class PersistentAsynchrony<T extends Serializable> extends HashFilePersistent {

    private int capacity = Integer.MAX_VALUE;

    protected final BlockingQueue<T> sharedQueue = new LinkedBlockingQueue<T>(getCapacity());

    private final Thread thread = new Thread(runnable());
    private boolean alive = false;

    /**
     * 具体如何消费数据.
     *
     * @param data 数据
     */
    protected abstract void take(T data);

    /**
     * 当前队列的大小.
     */
    public int size() {
        return sharedQueue.size();
    }

    protected int getCapacity() {
        return capacity;
    }

    public void setCapacity(int capacity) {
        this.capacity = capacity;
    }

    /**
     * 加入异步操作.
     *
     * @param data 需要处理的数据. 数据应实现{@link java.io.Serializable 序列化}接口
     */
    public void put(T data) {
        if (!alive) {
            throw new IllegalStateException("系统已关闭，不再接收消息：" + data);
        }
        try {
            sharedQueue.put(data);
            notifyNewPut();
        } catch (InterruptedException ex) {
            throw new RuntimeException("加入异步操作数据出错，丢失数据：" + data, ex);
        }
    }

    /**
     * 自定义线程名称.
     *
     * @return 返回"通用异步操作线程".
     */
    protected String getThreadName() {
        return "通用异步操作线程";
    }

    @Override
    protected void startup() {
        alive = true;
    }

    @Override
    protected void afterLoad() {
        ArrayList<T> data = deserialize("queue");
        if (!data.isEmpty()) {
            sharedQueue.addAll(data);
            notifyNewPut();
        }
    }

    @Override
    protected void beforeShutdown() {
        alive = false;
        thread.interrupt();
        ArrayList<T> data = new ArrayList<T>();
        sharedQueue.drainTo(data);
        addSerializable("queue", data);
    }

    private void notifyNewPut() {
        if (!thread.isAlive()) {
            thread.setDaemon(false);
            thread.setName(getThreadName());
            thread.start();
        }
    }

    protected Runnable runnable() {
        return new Runnable() {

            @Override
            public void run() {
                while (true) {
                    try {
                        T t = sharedQueue.take();
                        take(t);
                    } catch (InterruptedException ex) {
                        logger.debug("停止线程：{}", thread.getName());
                        break;
                    } catch (Exception ex) {
                        logger.error(thread.getName() + "忽略了错误:" + ex.getMessage(), ex);
                    }
                }
            }
        };
    }
}
